package cn.doitedu.sql;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class _15_LookupJoin {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(1000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);


        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        tenv.getConfig().set("table.exec.source.idle-timeout","1s");


        // 建表，映射kafka中的用户行为日志topic
        tenv.executeSql(
                " create table action_log(                                      "+
                        "   uid int,                                            "+
                        " 	event_id string,                                    "+
                        " 	properties map<string,string>,                      "+
                        " 	action_time bigint,                                 "+
                        " 	pt as proctime(),                                   "+
                        " 	rt as to_timestamp_ltz(action_time,3),              "+
                        " 	watermark for rt as rt                              "+
                        "  ) with (                                             "+
                        "    'connector' = 'kafka',                             "+
                        "    'topic' = 'action-log',                            "+
                        "    'properties.bootstrap.servers' = 'doitedu:9092',   "+
                        "    'properties.group.id' = 'g003',                    "+
                        "    'scan.startup.mode' = 'latest-offset',             "+
                        "    'value.format' = 'json',                           "+
                        "    'value.fields-include' = 'EXCEPT_KEY'              "+
                        " )                                                     "
        );


        tenv.executeSql(
                "CREATE TABLE students_mysql(    " +
                        "    id     int,         " +
                        "    name   string,      " +
                        "    gender string,      " +
                        "    age    int,         " +
                        "    salary float        " +
                        ") with (                " +
                        "    'connector' = 'jdbc',      " +
                        "    'url' = 'jdbc:mysql://doitedu:3306/doit46', " +
                        "    'table-name' = 'student', " +
                        "    'username' = 'root',       " +
                        "    'password' = 'root',       " +
                        "    'lookup.cache' = 'PARTIAL',       " +
                        "    'lookup.partial-cache.max-rows' = '100000',       " +
                        "    'lookup.partial-cache.expire-after-access' = '1 hour'       " +
                        ")");


        tenv.executeSql(
                        " SELECT                                                "+
                        "    a.uid,                                             "+
                        "    a.event_id,                                        "+
                        "    a.properties,                                      "+
                        "    a.action_time,                                     "+
                        "    b.name,                                            "+
                        "    b.gender,                                          "+
                        "    b.age,                                             "+
                        "    b.salary                                           "+
                        " FROM action_log as a                                  "+
                        " JOIN students_mysql FOR SYSTEM_TIME AS OF a.pt as b   "+
                        " ON a.uid = b.id                                       "
        ).print();



    }


}
